package edu.zut.cs.network.entity;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQBytesMessage;
import org.apache.activemq.command.ActiveMQObjectMessage;
import org.apache.activemq.command.ActiveMQTextMessage;
import org.apache.activemq.util.ByteSequence;
import org.junit.Test;

import edu.zut.cs.network.base.BaseMainCase;

public class QueueConsumer extends BaseMainCase {
	@Test
	public void testQueueConsumer() throws Exception {
		// 创建一个ConnectionFactory对象连接MQ服务器
		ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("yrl", "yrl",
				"tcp://192.168.57.138:61616");
		connectionFactory.setTrustAllPackages(true);//添加白名单
		// 创建一个连接对象
		Connection connection = connectionFactory.createConnection();
		// 开启连接
		connection.start();
		// 使用Connection对象创建一个Session对象，自动应答
		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		// 手动提交
//		 Session session = connection.createSession(false,
//		 Session.CLIENT_ACKNOWLEDGE);
		// 创建一个Destination对象。queue对象
		Queue queue = session.createQueue("sqlqueue2");
		// 使用Session对象创建一个消费者对象。
		MessageConsumer consumer = session.createConsumer(queue);
		// 接收消息
		consumer.setMessageListener(new MessageListener() {

			@Override
			public void onMessage(Message message) {
				// 打印结果
				
				ActiveMQBytesMessage bm = (ActiveMQBytesMessage) message;
				
				Field field = null;
				Class<?> clazz = bm.getClass();
				for (; clazz != Object.class; clazz = clazz.getSuperclass()) {
					try {
						field = clazz.getDeclaredField("content");
					} catch (Exception e) {
						// 这里甚么都不能抛出去。
						// 如果这里的异常打印或者往外抛，则就不会进入
					}
				}
				field.setAccessible(true);
				byte[] by = null;
				try {
					ByteSequence b = (ByteSequence) field.get(bm);
					by = b.getData();
				} catch (IllegalArgumentException e1) {
					e1.printStackTrace();
				} catch (IllegalAccessException e1) {
					e1.printStackTrace();
				}
				List<Book> l = unserializeForList(by);
				System.out.println(l);
			}

		});
		// 等待接收消息
		System.in.read();

		// 关闭资源
		consumer.close();
		session.close();
		connection.close();

	}

	public static <T> List<T> unserializeForList(byte[] in) {
		List<T> list = new ArrayList<T>();
		ByteArrayInputStream bis = null;
		ObjectInputStream is = null;
		try {
			if (in != null) {
				bis = new ByteArrayInputStream(in);
				is = new ObjectInputStream(bis);
				while (true) {
					T obj = (T) is.readObject();
					if (obj == null) {
						break;
					} else {
						list.add(obj);
					}
				}
				is.close();
				bis.close();
			}
		} catch (IOException e) {
			logger.warn("Caught IOException decoding %d bytes of data", in == null ? 0 : in.length, e);
		} catch (ClassNotFoundException e) {
			logger.warn("Caught CNFE decoding %d bytes of data", in == null ? 0 : in.length, e);
		}
		return list;
	}

}
